package com.atguigu.flink.aggregation;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class AggKeyBySum {

    public static void main(String[] args) throws Exception {

        // TODO 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // TODO 从固定端口获取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);

        // TODO 对流中数据进行类型转换
        SingleOutputStreamOperator<WaterSensor> wsDS = socketDS.map(
                new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String line) throws Exception {
                        String[] file = line.split(" ");
                        return new WaterSensor(file[0], Long.valueOf(file[1]), Integer.valueOf(file[2]));
                    }
                });

        // TODO 按照传感器id进行分组
        KeyedStream<WaterSensor, String> keyByDS = wsDS.keyBy(
                new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor ws) throws Exception {

                        return ws.id;
                    }
                });

        // TODO 统计
        SingleOutputStreamOperator<WaterSensor> sumDS = keyByDS.sum("vc");

        // TODO 打印
        sumDS.print();

        // TODO 提交作业
        env.execute();
    }
}
